package org.apm.data.delete;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.apm.data.bean.DataType;
import org.apm.data.bean.Parameter;
import org.apm.data.client.ClientFactory;
import org.apm.data.client.ScrollProcessor;
import org.apm.data.util.StringUtil;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.client.Client;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.joda.time.DateTime;

class DataDeleter {
    private static final Logger LOGGER = LogManager.getLogger(DataDeleter.class);

    private static final Client CLIENT = ClientFactory.getClient(DeleteConfig.CLUSTER_NAME, DeleteConfig.hosts);
    private static final BulkProcessor BULK_PROCESSOR = ClientFactory.getBulkProcessor(DeleteConfig.CLUSTER_NAME, DeleteConfig.hosts);

    private Parameter parameter;

    DataDeleter(final Parameter parameter) {
        this.parameter = parameter;
    }

    void process() {
        LOGGER.info("[Delete] Start deleting");

        if (parameter.isUser_operation()) {
            if (deleteUser(parameter.getMasterIp())) {
                LOGGER.info("[Delete] Delete user task has been accepted.");
            } else {
                LOGGER.info("[Delete] Delete user fail.");
            }
        } else if (parameter.isAll_type() || "*".equals(parameter.getType())) {
            LOGGER.info("[Delete] Delete all types.");
            for (DataType dataType : DataType.values()) {
                LOGGER.info("[Delete] Delete all types. Deleting " + dataType.type);
                delete(parameter.getMasterIp(), dataType.type, parameter.getFrom(), parameter.getTo());
            }
        } else if (parameter.isMulti_type()) {
            LOGGER.info("[Delete] Delete multi types.");
            for (String type : DeleteConfig.DELETE_MULTI_TYPE) {
                DataType dataType = DataType.dataType(type);
                LOGGER.info("[Delete] Delete multi types. Deleting " + dataType.type);
                delete(parameter.getMasterIp(), dataType.type, parameter.getFrom(), parameter.getTo());
            }
        } else {
            LOGGER.info("[Delete] Deleting " + parameter.getType());
            delete(parameter.getMasterIp(), parameter.getType(), parameter.getFrom(), parameter.getTo());

        }
    }

    private boolean deleteUser(final String masterIp) {
        String index = "apm2.0-" + masterIp + "-*-1";
        DeleteIndexResponse deleteIndexResponse = CLIENT.admin().indices()
                .delete(new DeleteIndexRequest().indices(index)).actionGet();

        return deleteIndexResponse.isAcknowledged();
    }

    private void delete(final String masterIp, final String type, final DateTime fromDate, final DateTime toDate) {
        DataType dataType = DataType.dataType(type);

        // 若没有时间字段,则直接返回
        if (StringUtil.isEmpty(dataType.timeField)) {
            LOGGER.info("[Delete] " + dataType.name() + " has no time field. Can't Delete by time range");
            return;
        }


        String index = "apm2.0-" + masterIp + "-" + type.toLowerCase() + "-1";
        String from = fromDate.toString(dataType.timeFormatter);
        String to = toDate.toString(dataType.timeFormatter);

        QueryBuilder query = QueryBuilders.boolQuery().filter(QueryBuilders.rangeQuery(dataType.timeField).from(from)
                .to(to).includeLower(true).includeUpper(false));

        DeleteScrollBean scrollBean = new DeleteScrollBean(CLIENT, BULK_PROCESSOR, index, query);
        ScrollProcessor.process(scrollBean);
        LOGGER.info("[Delete][" + index + "] finish deleting.");
    }
}
